use bytes::{BufMut, BytesMut};
use std::collections::HashMap;

use std::io::{Read, Write};
use std::net::TcpStream;
use crate::io::ConnectionStream;

use crate::error::{Result, VerticaError};
use crate::query::{ColumnInfo, QueryResult};
use crate::types::Value;

/// 异步连接模块，提供异步方式连接数据库的功能
pub mod async_conn;
use md5::Md5;
use md5::Digest;
/// SSL mode configuration
/// SSL 连接模式配置枚举
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum SslMode {
    /// 禁用 SSL 连接
    Disable,
    /// 允许使用 SSL 连接，如果服务器支持则使用，否则使用非 SSL 连接
    Allow,
    /// 优先使用 SSL 连接，如果服务器支持则使用，否则使用非 SSL 连接
    Prefer,
    /// 强制使用 SSL 连接，如果服务器不支持则连接失败
    Require,
    /// 强制使用 SSL 连接，并验证服务器证书是否由受信任的 CA 签发
    VerifyCa,
    /// 强制使用 SSL 连接，验证服务器证书是否由受信任的 CA 签发，并验证服务器主机名与证书中的域名匹配
    VerifyFull,
}
/// Connection state
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum ConnectionState {
    /// 连接已断开
    Disconnected,
    /// 正在尝试连接到数据库
    Connecting,
    /// 正在与数据库进行握手
    Handshake,
    /// 正在进行身份验证
    Authenticating,
    /// 连接已就绪，可以执行查询操作
    Ready,
    /// 连接过程中出现错误
    Error,
}
/// Connection configuration
#[derive(Debug, Clone)]
pub struct ConnectionConfig {
    /// 数据库服务器的主机名或 IP 地址
    pub host: String,
    /// 数据库服务器监听的端口号
    pub port: u16,
    /// 要连接的数据库名称
    pub database: String,
    /// 用于连接数据库的用户名
    pub username: String,
    /// 用于连接数据库的密码，可选字段
    pub password: Option<String>,
    /// SSL 连接模式配置
    pub ssl_mode: SslMode,
    /// SSL 客户端证书文件路径，可选字段
    pub ssl_cert: Option<String>,
    /// SSL 客户端私钥文件路径，可选字段
    pub ssl_key: Option<String>,
    /// SSL 根证书文件路径，可选字段
    pub ssl_root_cert: Option<String>,
    /// 连接数据库的超时时间
    pub connect_timeout: std::time::Duration,
    /// 读取数据的超时时间
    pub read_timeout: std::time::Duration,
    /// 写入数据的超时时间
    pub write_timeout: std::time::Duration,
    /// 备用服务器列表，每个元素为 "主机名:端口号" 格式的字符串，可选字段
    pub backup_servers: Option<Vec<String>>,
}

impl ConnectionConfig {
    /// Create a new configuration from URL
    pub fn from_url(url: &str) -> Result<Self> {
        let url = url::Url::parse(url)?;
        
        let host = url.host_str()
            .ok_or_else(|| VerticaError::Connection("Missing host in URL".to_string()))?
            .to_string();
        
        let port = url.port().unwrap_or(5433);
        let database = url.path().trim_start_matches('/').to_string();
        let username = url.username().to_string();
        let password = url.password().map(|p| p.to_string());
        
        let ssl_mode = match url.query_pairs()
            .find(|(k, _)| k == "sslmode")
            .map(|(_, v)| v.to_string())
            .as_deref()
        {
            Some("disable") => SslMode::Disable,
            Some("allow") => SslMode::Allow,
            Some("prefer") => SslMode::Prefer,
            Some("require") => SslMode::Require,
            Some("verify-ca") => SslMode::VerifyCa,
            Some("verify-full") => SslMode::VerifyFull,
            _ => SslMode::Prefer,
        };

        // Parse backup servers from query parameters
        let backup_servers = url.query_pairs()
            .find(|(k, _)| k == "backup_servers")
            .map(|(_, v)| {
                v.split(',')
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty())
                    .collect()
            });
        
        Ok(Self {
            host,
            port,
            database,
            username,
            password,
            ssl_mode,
            ssl_cert: None,
            ssl_key: None,
            ssl_root_cert: None,
            connect_timeout: std::time::Duration::from_secs(30),
            read_timeout: std::time::Duration::from_secs(30),
            write_timeout: std::time::Duration::from_secs(30),
            backup_servers,
        })
    }
}

/// Database connection
/// Database connection
pub struct Connection {
    stream: Option<ConnectionStream>,
    config: ConnectionConfig,
    state: ConnectionState,
    parameters: HashMap<String, String>,
    backend_pid: Option<u32>,
    secret_key: Option<u32>,
}

impl Connection {
    /// Create a new connection
    pub fn new(config: ConnectionConfig) -> Result<Self> {
        let mut conn = Self {
            stream: None,
            config,
            state: ConnectionState::Disconnected,
            parameters: HashMap::new(),
            backend_pid: None,
            secret_key: None,
        };
        conn.connect()?;
        Ok(conn)
    }

    /// Create a new connection from URL string
    pub fn new_from_url(url: &str) -> Result<Self> {
        let config = ConnectionConfig::from_url(url)?;
        Self::new(config)
    }

    /// Connect to the database - following vertica-sql-go's establishSocketConnection pattern
    fn connect(&mut self) -> Result<()> {
        log::debug!(
            "[CONNECT] Starting connection process to {}:{} as {}",
            self.config.host, self.config.port, self.config.username
        );
        self.state = ConnectionState::Connecting;

        // Build host list including backup servers
        let mut hosts = vec![format!("{}:{}", self.config.host, self.config.port)];
        if let Some(ref backup_servers) = self.config.backup_servers {
            hosts.extend(backup_servers.clone());
        }

        let mut _last_error = None;

        for host_port in &hosts {
            log::debug!("[CONNECT] Trying host: {}", host_port);

            match self.connect_to_host(host_port) {
                Ok(stream) => {
                    log::debug!("[CONNECT] Successfully connected to {}", host_port);
                    self.stream = Some(stream);
                    self.state = ConnectionState::Handshake;

                    return self.handshake().map_err(|e| {
                        log::debug!("[CONNECT] Handshake failed: {}", e);
                        e
                    });
                }
                Err(e) => {
                    log::debug!("[CONNECT] Failed to connect to {}: {}", host_port, e);
                    _last_error = Some(e);
                    continue;
                }
            }
        }

        let error_msg = format!("Failed to connect to any host. Last error: {}", 
                               _last_error.unwrap_or_else(|| VerticaError::Connection("Unknown error".to_string())));
        log::debug!("[CONNECT] {}", error_msg);
        Err(VerticaError::Connection(error_msg))
    }

    /// Connect to a specific host - implementing vertica-sql-go's connection pattern
    fn connect_to_host(&self, host_port: &str) -> Result<ConnectionStream> {
        let parts: Vec<&str> = host_port.split(':').collect();
        if parts.len() != 2 {
            return Err(VerticaError::Connection(format!("Invalid host:port format: {}", host_port)));
        }

        let host = parts[0];
        let port: u16 = parts[1].parse().map_err(|e| {
            VerticaError::Connection(format!("Invalid port number: {}", e))
        })?;

        // Resolve hostname to IP addresses
        use std::net::ToSocketAddrs;
        let addrs = (host, port).to_socket_addrs().map_err(|e| {
            VerticaError::Connection(format!("Failed to resolve host {}: {}", host, e))
        })?;

        let mut _last_error = None;

        for addr in addrs {
            log::debug!("[CONNECT] Trying IP address: {}", addr);

            match TcpStream::connect_timeout(&addr, self.config.connect_timeout) {
                Ok(stream) => {
                    // Set socket options like vertica-sql-go
                    stream.set_nodelay(true).map_err(|e| {
                        VerticaError::Connection(format!("Failed to set TCP_NODELAY: {}", e))
                    })?;

                    stream.set_read_timeout(Some(self.config.read_timeout)).map_err(|e| {
                        VerticaError::Connection(format!("Failed to set read timeout: {}", e))
                    })?;

                    stream.set_write_timeout(Some(self.config.write_timeout)).map_err(|e| {
                        VerticaError::Connection(format!("Failed to set write timeout: {}", e))
                    })?;

                    return Ok(ConnectionStream::Plain(stream));
                }
                Err(e) => {
                    log::debug!("[CONNECT] Failed to connect to {}: {}", addr, e);
                    _last_error = Some(e);
                    continue;
                }
            }
        }

        Err(VerticaError::Connection(format!("Failed to connect to {}: no address succeeded", host_port)))
    }

    /// Perform initial handshake
    fn handshake(&mut self) -> Result<()> {
        log::debug!("[HANDSHAKE] Starting handshake process");
        let mut buf = BytesMut::with_capacity(1024);
        
        // Startup message
        buf.put_i32(0); // Length placeholder
        buf.put_i32(196608); // Protocol version 3.0
        
        // Add parameters
        self.write_string(&mut buf, "user")?;
        self.write_string(&mut buf, &self.config.username)?;
        
        self.write_string(&mut buf, "database")?;
        self.write_string(&mut buf, &self.config.database)?;
        
        self.write_string(&mut buf, "client_encoding")?;
        self.write_string(&mut buf, "UTF8")?;
        
        self.write_string(&mut buf, "")?; // Terminator
        
        let len = buf.len() as i32;
        buf[0..4].copy_from_slice(&len.to_be_bytes());
        
        log::debug!("[HANDSHAKE] Sending startup message ({} bytes)", len);
        self.write_all(&buf).map_err(|e| {
            log::debug!("[HANDSHAKE] Failed to send startup message: {}", e);
            e
        })?;
        
        log::debug!("[HANDSHAKE] Startup message sent, starting authentication");
        self.authenticate()?;
        log::debug!("[HANDSHAKE] Handshake completed successfully");
        Ok(())
    }

    /// Authenticate with the server
    fn authenticate(&mut self) -> Result<()> {
        log::debug!("[AUTH] Starting authentication process");
        self.state = ConnectionState::Authenticating;
        
        let mut message_count = 0;
        loop {
            message_count += 1;
            log::debug!("[AUTH] Waiting for message #{}" , message_count);
            
            let mut msg = self.read_message().map_err(|e| {
                log::debug!("[AUTH] Failed to read message #{}: {}", message_count, e);
                e
            })?;
            
            log::debug!("[AUTH] Received message type: {}", msg.message_type as char);
            
            match msg.message_type {
                b'R' => {
                    let auth_type = msg.read_i32().map_err(|e| {
                        log::debug!("[AUTH] Failed to read auth type: {}", e);
                        e
                    })?;
                    log::debug!("[AUTH] Authentication type: {}", auth_type);
                    
                    match auth_type {
                        0 => { 
                            log::debug!("[AUTH] Authentication successful");
                        }
                        3 => { 
                            log::debug!("[AUTH] Clear text password required");
                            self.send_password_response()?;
                        }
                        5 => { 
                            log::debug!("[AUTH] MD5 password required");
                            let salt = msg.read_bytes(4)?;
                            log::debug!("[AUTH] Salt: {:02x?}", salt);
                            self.send_md5_password_response(&salt)?;
                        }
                        _ => {
                            let error = format!("Unsupported authentication type: {}", auth_type);
                            log::debug!("[AUTH] Error: {}", error);
                            return Err(VerticaError::Authentication(error));
                        }
                    }
                }
                b'K' => {
                    let pid = msg.read_u32()?;
                    let key = msg.read_u32()?;
                    self.backend_pid = Some(pid);
                    self.secret_key = Some(key);
                    log::debug!("[AUTH] Backend PID: {}, Secret key: {}", pid, key);
                }
                b'S' => {
                    let name = msg.read_string()?;
                    let value = msg.read_string()?;
                    self.parameters.insert(name.clone(), value.clone());
                    log::debug!("[AUTH] Parameter: {} = {}", name, value);
                }
                b'Z' => {
                    self.state = ConnectionState::Ready;
                    break;
                }
                _ => {
                    return Err(VerticaError::Connection(format!("Unexpected message type: {}", msg.message_type as char)));
                }
            }
        }
        
        Ok(())
    }

    /// Send clear text password response
    fn send_password_response(&mut self) -> Result<()> {
        let password = self.config.password.as_ref()
            .ok_or_else(|| VerticaError::Authentication("No password provided".to_string()))?;
        
        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'p');
        buf.put_i32(0); // Length placeholder
        self.write_string(&mut buf, password)?;
        
        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());
        
        log::debug!("[AUTH] Sending password response");
        self.write_all(&buf)?;
        Ok(())
    }

    /// Send MD5 password response
    fn send_md5_password_response(&mut self, salt: &[u8]) -> Result<()> {
        let password = self.config.password.as_ref()
            .ok_or_else(|| VerticaError::Authentication("No password provided".to_string()))?;
        
        // MD5 hash: md5(md5(password + username) + salt)
        let mut ctx = Md5::new();
        ctx.update(password.as_bytes());
        ctx.update(self.config.username.as_bytes());
        let hash1 = ctx.finalize();
        
        let mut ctx = Md5::new();
        ctx.update(format!("{:x}", hash1).as_bytes());
        ctx.update(salt);
        let hash2 = ctx.finalize();
        
        let password_hash = format!("md5{:x}", hash2);
        
        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'p');
        buf.put_i32(0); // Length placeholder
        self.write_string(&mut buf, &password_hash)?;
        
        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());
        
        log::debug!("[AUTH] Sending MD5 password response");
        self.write_all(&buf)?;
        Ok(())
    }

    /// Execute a simple query
    pub fn simple_query(&mut self, query: &str) -> Result<QueryResult> {
        if self.state != ConnectionState::Ready {
            return Err(VerticaError::Connection("Connection not ready".to_string()));
        }

        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'Q');
        buf.put_i32(0); // Length placeholder
        self.write_string(&mut buf, query)?;
        
        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());
        
        self.write_all(&buf)?;
        self.process_query_result()
    }
    /// Prepare a SQL statement
    pub fn prepare(&mut self, stmt_id: &str, query: &str) -> Result<()> {
        if self.state != ConnectionState::Ready {
            return Err(VerticaError::Connection("Connection not ready".to_string()));
        }

        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'P');
        buf.put_i32(0); // 长度占位符
        self.write_string(&mut buf, stmt_id)?;
        self.write_string(&mut buf, query)?;
        buf.put_i16(0); // 参数格式代码数量

        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());

        log::debug!("[PREPARE] Sending prepare statement request for stmt_id: {}", stmt_id);
        self.write_all(&buf)?;

        // 等待服务器响应
        loop {
            let mut msg = self.read_message()?;
            
            match msg.message_type {
                b'1' => {
                    // Parse Complete
                    log::debug!("[PREPARE] Prepare statement '{}' completed successfully", stmt_id);
                    break;
                }
                b'E' => {
                    // Error response
                    let fields = msg.read_string()?;
                    return Err(VerticaError::Query(fields));
                }
                _ => {
                    continue;
                }
            }
        }

        Ok(())
    }
    /// Execute a prepared query with parameters
    pub fn execute_prepared(&mut self, stmt_id: &str, params: &[Value]) -> Result<QueryResult> {
        if self.state != ConnectionState::Ready {
            return Err(VerticaError::Connection("Connection not ready".to_string()));
        }

        // 创建消息缓冲区
        let mut buf = BytesMut::with_capacity(1024);
        
        // 发送绑定消息
        buf.put_u8(b'B');
        buf.put_i32(0); // 长度占位符
        self.write_string(&mut buf, "")?; // 目标门户名称，空字符串表示默认
        self.write_string(&mut buf, stmt_id)?; // 准备语句ID
        
        // 绑定参数格式代码，这里假设所有参数都使用二进制格式
        buf.put_i16(params.len() as i16); // 参数格式代码数量
        for _ in params.iter() {
            buf.put_i16(1); // 1 表示二进制格式
        }
        
        // 绑定参数值
        buf.put_i16(params.len() as i16); // 参数值数量
        for param in params.iter() {
            match param {
                Value::Null => {
                    buf.put_i32(-1); // 表示 NULL 值
                }
                Value::String(s) => {
                    let bytes = s.as_bytes();
                    buf.put_i32(bytes.len() as i32);
                    buf.extend_from_slice(bytes);
                }
                Value::Int(i) => {
                    buf.put_i32(4);
                    buf.put_i32(*i);
                }
                Value::BigInt(i) => {
                    buf.put_i32(8);
                    buf.put_i64(*i);
                }
                Value::Float(f) => {
                    buf.put_i32(4);
                    buf.put_f32(*f);
                }
                Value::Double(f) => {
                    buf.put_i32(8);
                    buf.put_f64(*f);
                }
                Value::Boolean(b) => {
                    buf.put_i32(1);
                    buf.put_u8(*b as u8);
                }
                Value::Date(d) => {
                    let s = d.format("%Y-%m-%d").to_string();
                    let bytes = s.as_bytes();
                    buf.put_i32(bytes.len() as i32);
                    buf.extend_from_slice(bytes);
                }
                Value::Time(t) => {
                    let s = t.format("%H:%M:%S").to_string();
                    let bytes = s.as_bytes();
                    buf.put_i32(bytes.len() as i32);
                    buf.extend_from_slice(bytes);
                }
                Value::Timestamp(ts) => {
                    let s = ts.format("%Y-%m-%d %H:%M:%S").to_string();
                    let bytes = s.as_bytes();
                    buf.put_i32(bytes.len() as i32);
                    buf.extend_from_slice(bytes);
                }
                Value::Binary(b) => {
                    buf.put_i32(b.len() as i32);
                    buf.extend_from_slice(b);
                }
                Value::Uuid(u) => {
                    let s = u.to_string();
                    let _s_bytes = s.as_bytes();
                    buf.put_i32(s.len() as i32);
                    buf.extend_from_slice(s.as_bytes());
                }
                Value::Decimal(d) => {
                    let s = d.to_string();
                    let bytes = s.as_bytes();
                    buf.put_i32(bytes.len() as i32);
                    buf.extend_from_slice(bytes);
                }
            }
        }
        
        // 结果格式代码，这里假设所有结果都使用二进制格式
        buf.put_i16(0); // 结果格式代码数量
        
        // 更新消息长度
        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());
        
        log::debug!("[EXECUTE] Sending bind message for stmt_id: {}", stmt_id);
        self.write_all(&buf)?;
        
        // 发送执行消息
        let mut exec_buf = BytesMut::with_capacity(1024);
        exec_buf.put_u8(b'E');
        exec_buf.put_i32(0); // 长度占位符
        self.write_string(&mut exec_buf, "")?; // 目标门户名称，空字符串表示默认
        exec_buf.put_i32(0); // 返回行数，0 表示返回所有行
        
        // 更新执行消息长度
        let exec_len = exec_buf.len() as i32;
        exec_buf[1..5].copy_from_slice(&(exec_len - 1).to_be_bytes());
        
        log::debug!("[EXECUTE] Sending execute message");
        self.write_all(&exec_buf)?;
        
        // 发送同步消息
        let mut sync_buf = BytesMut::with_capacity(1024);
        sync_buf.put_u8(b'S');
        sync_buf.put_i32(4); // 消息长度为 4
        
        log::debug!("[EXECUTE] Sending sync message");
        self.write_all(&sync_buf)?;
        
        // 处理查询结果
        self.process_query_result()
    }

    /// Process query results
    fn process_query_result(&mut self) -> Result<QueryResult> {
        let mut rows = Vec::new();
        let mut columns = Vec::new();
        let mut affected_rows = 0;
        let mut command_tag = None;

        loop {
            let mut msg = self.read_message()?;
            
            match msg.message_type {
                b'T' => {
                    // Row description
                    let field_count = msg.read_i16()?;
                    for _ in 0..field_count {
                        let name = msg.read_string()?;
                        let _table_oid = msg.read_u32()?;
                        let _column_attr_number = msg.read_i16()?;
                        let data_type = msg.read_u32()?;
                        let _data_type_size = msg.read_i16()?;
                        let _type_modifier = msg.read_i32()?;
                        let _format_code = msg.read_i16()?;
                        
                        columns.push(ColumnInfo {
                            name,
                            data_type,
                            nullable: true, // TODO: Parse from type modifier
                            precision: None,
                            scale: None,
                        });
                    }
                }
                b'D' => {
                    // Data row
                    let column_count = msg.read_i16()?;
                    let mut row = Vec::new();
                    
                    for i in 0..column_count {
                        let len = msg.read_i32()?;
                        if len == -1 {
                            row.push(Value::Null);
                        } else {
                            let bytes = msg.read_bytes(len as usize)?;
                            
                            // 根据列类型转换数据
                            let value = if let Some(column_info) = columns.get(i as usize) {
                                match column_info.data_type {
                                    // 字符串类型 - 直接转换为字符串
                                    crate::types::vertica_types::VARCHAR |
                                    crate::types::vertica_types::CHAR |
                                    crate::types::vertica_types::TEXT => {
                                        String::from_utf8(bytes.clone())
                                            .map(Value::String)
                                            .unwrap_or_else(|_| Value::Binary(bytes.clone()))
                                    },
                                    
                                    // UUID类型 - 解析为 Uuid
                                    crate::types::vertica_types::UUID => {
                                        if bytes.len() >= 16 {
                                            let uuid_bytes: [u8; 16] = bytes[..16].try_into()
                                                .unwrap_or([0u8; 16]);
                                            Value::Uuid(uuid::Uuid::from_bytes(uuid_bytes))
                                        } else {
                                            Value::Binary(bytes.clone())
                                        }
                                    },
                                    // 整数类型
                                    crate::types::vertica_types::INTEGER => {
                                        match bytes.len() {
                                            4 => {
                                                // 标准4字节整数
                                                let value = i32::from_be_bytes([
                                                    bytes[0], bytes[1], bytes[2], bytes[3]
                                                ]);
                                                Value::Int(value)
                                            },
                                            1 => {
                                                // 单字节整数 (常见于小整数值)
                                                let value = i8::from_be_bytes([bytes[0]]) as i32;
                                                Value::Int(value)
                                            },
                                            2 => {
                                                // 双字节整数
                                                let value = i16::from_be_bytes([bytes[0], bytes[1]]) as i32;
                                                Value::Int(value)
                                            },
                                            _ => {
                                                // 尝试字符串格式解析（适用于任何长度的字符串表示的数字）
                                                let s = String::from_utf8(bytes.clone())
                                                    .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                                match s.trim().parse::<i32>() {
                                                    Ok(num) => Value::Int(num),
                                                    Err(_) => {
                                                        // 如果解析失败，尝试解析为i64然后转换
                                                        match s.trim().parse::<i64>() {
                                                            Ok(num) => Value::Int(num as i32),
                                                            Err(_) => Value::String(s)
                                                        }
                                                    }
                                                }
                                            }
                                        }
                                    },
                                    // 大整数类型
                                    crate::types::vertica_types::BIGINT => {
                                        match bytes.len() {
                                            8 => {
                                                let value = i64::from_be_bytes([
                                                    bytes[0], bytes[1], bytes[2], bytes[3],
                                                    bytes[4], bytes[5], bytes[6], bytes[7]
                                                ]);
                                                Value::BigInt(value)
                                            },
                                            4 => {
                                                // 4字节整数可以转换为i64
                                                let value = i32::from_be_bytes([
                                                    bytes[0], bytes[1], bytes[2], bytes[3]
                                                ]) as i64;
                                                Value::BigInt(value)
                                            },
                                            _ => {
                                                // 尝试字符串格式解析
                                                let s = String::from_utf8(bytes.clone())
                                                    .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                                match s.trim().parse::<i64>() {
                                                    Ok(num) => Value::BigInt(num),
                                                    Err(_) => Value::String(s)
                                                }
                                            }
                                        }
                                    },
                                    // 浮点类型（4字节）
                                    crate::types::vertica_types::FLOAT => {
                                        if bytes.len() == 4 {
                                            let value = f32::from_be_bytes([
                                                bytes[0], bytes[1], bytes[2], bytes[3]
                                            ]);
                                            Value::Float(value)
                                        } else if bytes.len() == 8 {
                                            let value = f64::from_be_bytes([
                                                bytes[0], bytes[1], bytes[2], bytes[3],
                                                bytes[4], bytes[5], bytes[6], bytes[7]
                                            ]);
                                            Value::Double(value)
                                        } else {
                                            Value::Binary(bytes.clone())
                                        }
                                    },
                                    // 布尔类型
                                    crate::types::vertica_types::BOOLEAN => {
                                        if !bytes.is_empty() {
                                            Value::Boolean(bytes[0] != 0)
                                        } else {
                                            Value::Binary(bytes.clone())
                                        }
                                    },
                                    // NUMERIC类型（高精度数值）
                                    crate::types::vertica_types::NUMERIC => {
                                        // NUMERIC类型通常以字符串形式传输
                                        String::from_utf8(bytes.clone())
                                            .map(|s| {
                                                // 尝试解析为Decimal，如果不能解析则保持字符串
                                                s.parse::<rust_decimal::Decimal>()
                                                    .map(Value::Decimal)
                                                    .unwrap_or_else(|_| Value::String(s))
                                            })
                                            .unwrap_or_else(|_| Value::Binary(bytes.clone()))
                                    },
                                    // DATE类型（日期）
                                    crate::types::vertica_types::DATE => {
                                        let s = String::from_utf8(bytes.clone())
                                            .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                        match chrono::NaiveDate::parse_from_str(&s, "%Y-%m-%d") {
                                            Ok(date) => Value::Date(date),
                                            Err(_) => Value::String(s), // Fallback if parsing fails
                                        }
                                    },
                                    // TIME类型（时间）
                                    crate::types::vertica_types::TIME => {
                                        let s = String::from_utf8(bytes.clone())
                                            .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                        match chrono::NaiveTime::parse_from_str(&s, "%H:%M:%S") {
                                            Ok(time) => Value::Time(time),
                                            Err(_) => Value::String(s), // Fallback if parsing fails
                                        }
                                    },
                                    // TIMESTAMP类型（时间戳）
                                    crate::types::vertica_types::TIMESTAMP => {
                                        let s = String::from_utf8(bytes.clone())
                                            .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                        // 尝试多种时间格式
                                        let formats = [
                                            "%Y-%m-%d %H:%M:%S",
                                            "%Y-%m-%d %H:%M:%S%.f",
                                            "%Y-%m-%d %H:%M",
                                            "%Y-%m-%d"
                                        ];
                                        
                                        let mut result = Value::String(s.clone());
                                        for fmt in formats.iter() {
                                            if let Ok(timestamp) = chrono::NaiveDateTime::parse_from_str(&s.trim(), fmt) {
                                                result = Value::Timestamp(timestamp);
                                                break;
                                            }
                                        }
                                        
                                        // 如果还没解析成功，尝试解析为日期再转换为时间戳
                                        if let Value::String(_) = result {
                                            if let Ok(date) = chrono::NaiveDate::parse_from_str(&s.trim(), "%Y-%m-%d") {
                                                result = Value::Timestamp(date.and_hms_opt(0, 0, 0).unwrap());
                                            }
                                        }
                                        
                                        result
                                    },
                                    // TIMESTAMPTZ类型（带时区的时间戳）
                                    crate::types::vertica_types::TIMESTAMPTZ => {
                                        let s = String::from_utf8(bytes.clone())
                                            .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                        // 尝试多种时间格式
                                        let formats = [
                                            "%Y-%m-%d %H:%M:%S",
                                            "%Y-%m-%d %H:%M:%S%.f",
                                            "%Y-%m-%d %H:%M",
                                            "%Y-%m-%d"
                                        ];
                                        
                                        let mut result = Value::String(s.clone());
                                        for fmt in formats.iter() {
                                            if let Ok(timestamp) = chrono::NaiveDateTime::parse_from_str(&s.trim(), fmt) {
                                                result = Value::Timestamp(timestamp);
                                                break;
                                            }
                                        }
                                        
                                        // 如果还没解析成功，尝试解析为日期再转换为时间戳
                                        if let Value::String(_) = result {
                                            if let Ok(date) = chrono::NaiveDate::parse_from_str(&s.trim(), "%Y-%m-%d") {
                                                result = Value::Timestamp(date.and_hms_opt(0, 0, 0).unwrap());
                                            }
                                        }
                                        
                                        result
                                    },
                                    // TIMETZ类型（带时区的时间）
                                    crate::types::vertica_types::TIMETZ => {
                                        let s = String::from_utf8(bytes.clone())
                                            .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).to_string());
                                        let s = s.trim();
                                        
                                        // 为了调试，打印原始值
                                        eprintln!("DEBUG: TIMETZ raw value: [{}]", s);
                                        
                                        // 尝试多种格式解析
                                        let formats = ["%H:%M:%S%.f", "%H:%M:%S", "%H:%M:%S%.f%z", "%H:%M:%S%z"];
                                        let mut time = None;
                                        
                                        // 提取时间部分（去掉时区）
                                        let time_part = if let Some(pos) = s.find(|c: char| c == '+' || c == '-') {
                                            &s[..pos]
                                        } else {
                                            &s
                                        };
                                        
                                        eprintln!("DEBUG: Time part: [{}]", time_part);
                                        
                                        for fmt in formats.iter() {
                                            eprintln!("DEBUG: Trying format: {}", fmt);
                                            if let Ok(t) = chrono::NaiveTime::parse_from_str(time_part, fmt) {
                                                eprintln!("DEBUG: Successfully parsed with {}: {:?}", fmt, t);
                                                time = Some(t);
                                                break;
                                            }
                                        }
                                        
                                        if let Some(t) = time {
                                            Value::Time(t)
                                        } else {
                                            eprintln!("DEBUG: Failed to parse, using String");
                                            Value::String(s.to_string())
                                        }
                                    },
                                    // INTERVAL类型（时间间隔）
                                    crate::types::vertica_types::INTERVAL | 
                                    crate::types::vertica_types::INTERVAL_YM => {
                                        String::from_utf8(bytes.clone())
                                            .map(|s| Value::String(s))
                                            .unwrap_or_else(|_| Value::Binary(bytes.clone()))
                                    },
                                    // LONG VARBINARY类型
                                    crate::types::vertica_types::LONG_VARBINARY => {
                                        Value::Binary(bytes.clone())
                                    },
                                    // BINARY类型（固定长度二进制）
                                    crate::types::vertica_types::BINARY => {
                                        Value::Binary(bytes.clone())
                                    },
                                    // 其他类型保持二进制格式
                                    _ => Value::Binary(bytes.clone()),
                                }
                            } else {
                                // 如果没有列信息，尝试UTF-8转换
                                String::from_utf8(bytes.clone())
                                    .map(Value::String)
                                    .unwrap_or_else(|_| Value::Binary(bytes.clone()))
                            };
                            row.push(value);
                        }
                    }
                    
                    rows.push(row);
                }
                b'C' => {
                    // Command complete
                    let tag = msg.read_string()?;
                    command_tag = Some(tag.clone());
                    
                    // Parse affected rows from command tag
                    static ROW_COUNT_REGEX: once_cell::sync::Lazy<regex::Regex> = once_cell::sync::Lazy::new(|| {
                        regex::Regex::new(r"(\d+)").expect("Failed to compile row count regex")
                    });
                    
                    if let Some(caps) = ROW_COUNT_REGEX.captures(&tag) {
                        if let Some(matched) = caps.get(1) {
                            affected_rows = matched.as_str().parse().unwrap_or(0);
                        }
                    }
                }
                b'Z' => {
                    // Ready for query
                    let _status = msg.read_u8()?;
                    break;
                }
                b'E' => {
                    // Error response
                    let fields = msg.read_string()?;
                    return Err(VerticaError::Query(fields));
                }
                _ => {
                    // Skip unknown messages
                    continue;
                }
            }
        }

        if rows.is_empty() {
            Ok(QueryResult::from_command(affected_rows, None, command_tag.unwrap_or_default()))
        } else {
            Ok(QueryResult::from_rows(rows, columns))
        }
    }

    /// Read a message from the server
    fn read_message(&mut self) -> Result<Message> {
        log::debug!("[READ] Waiting for message header...");
        let mut header = [0u8; 5];
        self.read_exact(&mut header).map_err(|e| {
            log::debug!("[READ] Failed to read message header: {}", e);
            e
        })?;
        
        let msg_type = header[0];
        let len = i32::from_be_bytes([header[1], header[2], header[3], header[4]]) - 4;
        
        log::debug!("[READ] Message type: {} ({}), payload length: {}", 
                 msg_type as char, msg_type, len);
        
        if len < 0 {
            return Err(VerticaError::Connection(format!("Invalid message length: {}", len)));
        }
        
        let mut payload = vec![0u8; len as usize];
        if len > 0 {
            log::debug!("[READ] Reading payload of {} bytes...", len);
            self.read_exact(&mut payload).map_err(|e| {
                log::debug!("[READ] Failed to read payload: {}", e);
                e
            })?;
            log::debug!("[READ] Payload read successfully");
        }
        
        Ok(Message::new(msg_type, payload))
    }

    /// Write a string to buffer
    fn write_string(&self, buf: &mut BytesMut, s: &str) -> Result<()> {
        buf.extend_from_slice(s.as_bytes());
        buf.put_u8(0);
        Ok(())
    }

    /// Write all bytes to stream
    fn write_all(&mut self, buf: &[u8]) -> Result<()> {
        log::debug!("[WRITE] Writing {} bytes to stream...", buf.len());
        if let Some(ref mut _stream) = self.stream {
            _stream.write_all(buf).map_err(|e| {
                log::debug!("[WRITE] Failed to write to stream: {}", e);
                e
            })?;
            _stream.flush().map_err(|e| {
                log::debug!("[WRITE] Failed to flush stream: {}", e);
                e
            })?;
            log::debug!("[WRITE] Successfully wrote {} bytes", buf.len());
            Ok(())
        } else {
            let error = "No active stream".to_string();
            log::debug!("[WRITE] Error: {}", error);
            Err(VerticaError::Connection(error))
        }
    }

    /// Read exact bytes from stream - following vertica-sql-go's readAll pattern
    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        let bytes_to_read = buf.len();
        log::debug!("[READ] Reading {} bytes from stream...", bytes_to_read);
        if let Some(ref mut _stream) = self.stream {
            let start_time = std::time::Instant::now();
            let mut read_index = 0;
            
            while read_index < bytes_to_read {
                match _stream.read(&mut buf[read_index..]) {
                    Ok(bytes_read) => {
                        if bytes_read == 0 {
                            // Connection closed by peer
                            let elapsed = start_time.elapsed();
                            log::debug!("[READ] Connection closed by peer after {:?}, read {}/{} bytes", 
                                       elapsed, read_index, bytes_to_read);
                            return Err(VerticaError::Connection("Connection closed by peer".to_string()));
                        }
                        read_index += bytes_read;
                        log::trace!("[READ] Read {}/{} bytes", read_index, bytes_to_read);
                    }
                    Err(e) => {
                        let elapsed = start_time.elapsed();
                        log::debug!("[READ] Failed to read bytes after {:?}: {}", elapsed, e);
                        return Err(VerticaError::Connection(e.to_string()));
                    }
                }
            }
            
            let elapsed = start_time.elapsed();
            log::debug!("[READ] Successfully read {} bytes in {:?}", bytes_to_read, elapsed);
            Ok(())
        } else {
            let error = "No active stream".to_string();
            log::debug!("[READ] Error: {}", error);
            Err(VerticaError::Connection(error))
        }
    }

    /// Get connection state
    pub fn state(&self) -> ConnectionState {
        self.state
    }

    /// Check if connection is ready
    pub fn is_ready(&self) -> bool {
        self.state == ConnectionState::Ready
    }

    /// Get server parameters
    pub fn parameters(&self) -> &HashMap<String, String> {
        &self.parameters
    }

    /// Close the connection
    pub fn close(&mut self) -> Result<()> {
        if let Some(ref mut _stream) = self.stream {
            // Send terminate message
            let mut buf = BytesMut::with_capacity(5);
            buf.put_u8(b'X');
            buf.put_i32(4);
            let _ = self.write_all(&buf);
        }
        
        self.stream = None;
        self.state = ConnectionState::Disconnected;
        Ok(())
    }

    /// Clone the connection (shallow clone)
    pub fn clone(&self) -> Self {
        Self {
            stream: None, // Don't clone the actual stream
            config: self.config.clone(),
            state: ConnectionState::Disconnected,
            parameters: self.parameters.clone(),
            backend_pid: self.backend_pid,
            secret_key: self.secret_key,
        }
    }
}

/// Message structure for protocol communication
struct Message {
    message_type: u8,
    payload: Vec<u8>,
    position: usize,
}

impl Message {
    fn new(message_type: u8, payload: Vec<u8>) -> Self {
        Self {
            message_type,
            payload,
            position: 0,
        }
    }

    fn read_u8(&mut self) -> Result<u8> {
        if self.position >= self.payload.len() {
            return Err(VerticaError::Connection("read_u8:Unexpected end of message".to_string()));
        }
        let val = self.payload[self.position];
        self.position += 1;
        Ok(val)
    }

    fn read_i16(&mut self) -> Result<i16> {
        if self.position + 2 > self.payload.len() {
            return Err(VerticaError::Connection("read_i16:Unexpected end of message".to_string()));
        }
        let val = i16::from_be_bytes([self.payload[self.position], self.payload[self.position + 1]]);
        self.position += 2;
        Ok(val)
    }

    fn read_i32(&mut self) -> Result<i32> {
        if self.position + 4 > self.payload.len() {
            return Err(VerticaError::Connection("read_i32:Unexpected end of message".to_string()));
        }
        let val = i32::from_be_bytes([
            self.payload[self.position],
            self.payload[self.position + 1],
            self.payload[self.position + 2],
            self.payload[self.position + 3],
        ]);
        self.position += 4;
        Ok(val)
    }

    fn read_u32(&mut self) -> Result<u32> {
        if self.position + 4 > self.payload.len() {
            return Err(VerticaError::Connection("read_u32:Unexpected end of message".to_string()));
        }
        let val = u32::from_be_bytes([
            self.payload[self.position],
            self.payload[self.position + 1],
            self.payload[self.position + 2],
            self.payload[self.position + 3],
        ]);
        self.position += 4;
        Ok(val)
    }

    fn read_string(&mut self) -> Result<String> {
        let null_pos = self.payload[self.position..]
            .iter()
            .position(|&b| b == 0)
            .ok_or_else(|| VerticaError::Connection("Unterminated string".to_string()))?;
        
        let s = String::from_utf8_lossy(&self.payload[self.position..self.position + null_pos]).to_string();
        self.position += null_pos + 1;
        Ok(s)
    }

    fn read_bytes(&mut self, len: usize) -> Result<Vec<u8>> {
        if self.position + len > self.payload.len() {
            return Err(VerticaError::Connection("read_bytes:Unexpected end of message".to_string()));
        }
        let bytes = self.payload[self.position..self.position + len].to_vec();
        self.position += len;
        Ok(bytes)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_connection_config() {
        let config = ConnectionConfig::from_url("vertica://user:pass@localhost:5433/db").unwrap();
        assert_eq!(config.host, "localhost");
        assert_eq!(config.port, 5433);
        assert_eq!(config.database, "db");
        assert_eq!(config.username, "user");
        assert_eq!(config.password, Some("pass".to_string()));
    }
}

use std::fmt;

impl fmt::Debug for Connection {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Connection")
            .field("config", &self.config)
            .field("state", &self.state)
            .field("parameters", &self.parameters)
            .field("backend_pid", &self.backend_pid)
            .field("secret_key", &self.secret_key)
            .finish_non_exhaustive()
    }
}